package com.example;

import com.google.common.collect.Lists;
import com.sun.javafx.binding.StringFormatter;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ConsumerDemo
{

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Consumer<String,String> consumer = new KafkaConsumer<String, String>(initConfig());

        consumer.subscribe(Lists.newArrayList("topic_Robot"));

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));

            if (records.isEmpty()){
                System.err.println("本次拉取没有数据");
                continue;
            }
            for (ConsumerRecord<String, String> record : records) {
                String key = record.key();
                String value = record.value();
                int partition = record.partition();
                System.err.println("获取到数据");
                System.err.println("partition="+partition);
                System.err.println("key="+key);
                System.err.println("value="+value);
                System.err.println("======================================================");
            }
        }




    }


    private static Properties initConfig(){
        Properties properties = new Properties();
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.199.200:9094,192.168.199.200:9095,192.168.199.200:9096");
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.201.83.128:9094");
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.201.81.64:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, ConsumerDemo.class.getName());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return properties;
    }
}
